#include <jni.h>
#include <stdio.h>
#include <errno.h>
//#include "ktcpjni.h"

#include "caevent.h"
#include "casock.h"
#include "cathread.h"
#include "calog.h"

#include "CGetPutBuf.h"
#include "renderscr.h"
#include "capturesr.h"
#include "udpxbuf.h"

#define TEST_PORT 9010

#define SCX (&gSendCtx)
es_t gSendCtx;

#define RCX (&gReadCtx)
es_t gReadCtx;

#define CHUNK_SIZE 65000


CGetPutBuf* gGPBuf;



pthread_t tid_tx, tid_rx;

#if 0
typedef struct {
	char ip[32];
	int port;
	int imgw, imgh;
} xarg;
#endif
xarg txinfo, rxinfo;


typedef struct 
{
	int status; // 0== idle
	int size;
	int cur_pos;
	unsigned char* buf;
} txstatus_t;

static txstatus_t gTx;

void* rx_thread(void* arg);
void* tx_thread(void* arg);



int SendStart (JNIEnv *env,  jstring jipaddr, jint port, int w, int h, int trans)
{
	logprt("== nv send start.........tid_tx=%d\n", tid_tx);
	if(tid_tx)
		return -1;
	const char* ip = env->GetStringUTFChars(jipaddr, 0);
	strcpy(txinfo.ip, ip);
	txinfo.port = port;
	txinfo.trans = trans;
	env->ReleaseStringUTFChars(jipaddr, ip);
	logprt("== nv send start, ip=%s, port=%d, w=%d, h=%d\n", txinfo.ip, txinfo.port, w, h);

	txinfo.imgw = w;
	txinfo.imgh =h ;
	
	es_thread_start(&tid_tx, tx_thread, NULL, SCX);
	logprt("    ---- tx thread rady.......\n");
	return 0;
}



/*
 * Class:     com_example_tcpsr_ktcp
 * Method:    nvRecvStart
 * Signature: (I)I
 */
int RecvStart(JNIEnv *env, jstring jipaddr, jint port, int w, int h, int trans)
{
	logprt("== nv recv start.........tid_tx=%d, w=%d, h=%d\n", tid_rx, w, h);

	if(tid_rx)
		return -1;
	const char* ip = env->GetStringUTFChars(jipaddr, 0);
	strcpy(rxinfo.ip, ip);
	rxinfo.port = port;
	rxinfo.trans = trans;
	env->ReleaseStringUTFChars(jipaddr, ip);
	rxinfo.imgw = w;
	rxinfo.imgh = h;
	logprt("== recv start, ip=%s, port=%d\n", rxinfo.ip, rxinfo.port);
	es_thread_start(&tid_rx, rx_thread, NULL, RCX);
	return 0;
}


void SendStop(void)
{
	if(!tid_tx)
		return;
	es_exit(SCX);
	es_thread_stop(tid_tx);
	tid_tx = 0;
}

void RecvStop(void)
{
	if(!tid_rx)
		return;
	es_exit(RCX);
	es_thread_stop(tid_rx);
	tid_rx = 0;
}



#define SOCKID_TX_LISTEN 1
#define SOCKID_TX_SEND 2

int TxMsgProc(es_t* psys, uint16_t msgid, uint16_t p1, uint32_t p2)
{
	static sock_t sendsock, *psock;
	static sock_t listen_sock, *plsock;
	static char* send_buf;
	static int req_wsize, cur_size;
	static int exp_size;
	static int wmode;
	//static int break_time;
	static int skip_frame = 0;
#define TIMER_SEND 1

	int ret;	
	if(msgid == EM_INIT)
	{
		logprt("== TxMsgProc init.....\n");
		req_wsize = cur_size = 0;
		//break_time = 0;
		
		open_scr_encoder();
		wmode = 0;
		psock = &sendsock;
		plsock = &listen_sock;
		start_cap_thread(txinfo.imgw, txinfo.imgh, PIXEL_BYTE);		
		es_sock_open(psys, plsock, KSOCK_TYPE_TCP, NULL, (void*)SOCKID_TX_LISTEN);
		es_tcp_listen(plsock, "0.0.0.0", txinfo.port);
		
	}
	else if(msgid == UM_CONNECT)
	{
		ret = es_connect(psock, txinfo.ip, txinfo.port);
		logprt("== connecting... ip=%s, port=%d, ret=%d\n", txinfo.ip, txinfo.port, ret);
		if(ret !=0 && errno != EINPROGRESS)
		{
			logprt("### connect error.......ret=%d, errno=%d\n", ret, errno);
			es_exit(psys);
		}

	}
	else if(msgid == EM_TIMER)
	{
		void* img;
		unsigned long jpegSize;
		if(p1==TIMER_SEND ) {
			if(wmode != 0)
			{
				logprt("## postphone send frame ....\n");
				if(psock->status == SOCK_STATUS_DISCONNECTED)
				{
					logprt("####### unexpected status...disconnected....\n");
					assert(0);
				}	
				return 0;
			}

			if(skip_frame >0 )
			{
				logprt("== skip frame, skip=%d\n", skip_frame);
				skip_frame--;				
				return 0;
			}
			
			//logprt("== start write, \n");

			img = gCapImgBuf->get_fill();
			if(img)
			{
				//unsigned int t1, t2;
				//t1 = es_mtime(psys);
				send_buf = (char*)encode_scr_jpeg(img, &jpegSize, txinfo.imgw, txinfo.imgh);
				//t2 = es_mtime(psys);
				//logprt("== enc time=%d\n", t2-t1);
				gCapImgBuf->put_empty(img);
			}	
			else
			{
				//logprt("++ empty cap img......\n");
				return 0;
			}	
			
			if(!send_buf)
			{
				logprt("### Error: encoding.....\n");
				return 0;
			}
			exp_size = jpegSize + PKT_HDR_SIZE;
			ret = es_write(psock, send_buf, exp_size);
			//ret = es_write(psock, send_buf, 1480); // test
			//logprt("== es write=%d, jpegSize=%d, send_size=%d\n", ret, jpegSize, ((int*)send_buf)[0]);
			if(ret != exp_size) {
				if(ret>0)
					cur_size += ret;
				logprt("## write error, ret=%d, errno=%d\n", ret, errno);
				//break_time = 150;
				skip_frame = 2;
				//gCapImgBuf->reset();
				//es_change(evt_t * pevt,uint32_t event)
				es_set_sock_event(psock, EPOLLIN|EPOLLOUT);
				wmode = 1;		
			}
			else
			{
				#if 0
				if(break_time>0)
				{
					break_time -= 10;
					if(break_time<0)
						break_time = 0;
					adjust_cap_timer(break_time);	
				}
				#endif
				wmode = 0;
			}
		}
	}
	else if(msgid == EM_SOCKET)
	{

		if(p2 == SOCKID_TX_LISTEN)
		{
			if(psock->status != SOCK_STATUS_DISCONNECTED)
			{
				logprt("== child already connected... rejecting...\n");
				es_tcp_reject(plsock);
				return 0;
			}
			else 
			{
				logprt("== send socket accepted....\n");
				cur_size = req_wsize = 0;
				wmode = 0;
				es_tcp_accept(plsock, psock, NULL, (void*)SOCKID_TX_SEND);
				start_capture_samp();
				es_set_timer(psys, TIMER_SEND, CAP_PERIOD, NULL);
			}
		}
		else if(p2 == SOCKID_TX_SEND)
		{
			if(p1 == SOCK_EVENT_WRITE)
			{
				logprt("== write event...\n");
				if(1)//wmode != 0)
				{
					ret = es_write(psock, send_buf+cur_size, exp_size-cur_size);
					//ret = es_write(psock, send_buf+cur_size, (exp_size-cur_size)>1480 ? 1480:(exp_size-cur_size)); //test
					if(ret>0)
					{
						cur_size += ret;
						if(cur_size == exp_size)
						{
							logprt("== write end...,cur_size=%d\n", cur_size);
							es_set_sock_event(psock, EPOLLIN);
							cur_size = req_wsize = 0;
							wmode = 0;
						}
					}
					/*
					else { // test 0805
						logprt("== write end...,cur_size=%d\n", cur_size);
						es_set_sock_event(psock, EPOLLIN|EPOLLOUT);
					}*/
				}	
			}
			else if(p1 == SOCK_EVENT_DISCONNECTED)
			{
				logprt("== tx disconnected......\n");
				es_kill_timer(psys, TIMER_SEND);
				stop_capture_samp();
				es_tcp_close(psock);
			}
			else if(p1 == SOCK_EVENT_READ)
			{	
				char buf[128];
				int rdcnt = es_read(psock, buf, sizeof(buf));
				// may be disconnected......
				return rdcnt;
			}
		}
	}
	else if(msgid == EM_CLOSE)
	{
		es_kill_timer(psys, TIMER_SEND);
		stop_cap_thread();
		es_tcp_close( psock);
		es_tcp_close(plsock);
		close_scr_encoder();
		
	}
	
	return 0;
}



int TxUDPMsgProc(es_t* psys, uint16_t msgid, uint16_t p1, uint32_t p2)
{
	//static int break_time;
	static int skip_frame = 0;
	static CUdpXBuf* udpx;
	unsigned char* send_buf;
#define TIMER_SEND 1

	int ret;	
	if(msgid == EM_INIT)
	{
		logprt("== TxUDPMsgProc init.....\n");
		
		open_scr_encoder();
		start_cap_thread(txinfo.imgw, txinfo.imgh, PIXEL_BYTE);
		udpx = new CUdpXBuf(psys, 100*1024, 0, NULL);
		ret = udpx->connect(txinfo.ip, txinfo.port);
		if(ret !=0)
		{
			logprt("### connect error.......ret=%d, errno=%d\n", ret, errno);
			es_exit(psys);
		}
		es_set_timer(psys, TIMER_SEND, CAP_PERIOD, NULL);
		start_capture_samp();
	}
	/*
	else if(msgid == UM_CONNECT)
	{
		ret = udpx->connect(txinfo.ip, txinfo.port);
		logprt("== connecting... ip=%s, port=%d, ret=%d\n", txinfo.ip, txinfo.port, ret);
		if(ret !=0 && errno != EINPROGRESS)
		{
			logprt("### connect error.......ret=%d, errno=%d\n", ret, errno);
			es_exit(psys);
		}
		es_set_timer(psys, TIMER_SEND, CAP_PERIOD, NULL);
	}*/
	else if(msgid == EM_TIMER)
	{
		void* img;
		unsigned long jpegSize;
		if(p1==TIMER_SEND ) {
			if(skip_frame >0 )
			{
				logprt("== skip frame, skip=%d\n", skip_frame);
				skip_frame--;				
				return 0;
			}

			if(udpx->status() != 0)
			{
				logprt("++ skip for transmission incomplete...\n");
				return 0;
			}
			
			
			//logprt("== start write, \n");

			img = gCapImgBuf->get_fill();
			if(img)
			{
				//unsigned int t1, t2;
				//t1 = es_mtime(psys);
				send_buf = encode_raw_jpeg(img, &jpegSize, txinfo.imgw, txinfo.imgh);
				//t2 = es_mtime(psys);
				//logprt("== enc time=%d\n", t2-t1);
				gCapImgBuf->put_empty(img);
			}	
			else
			{
				logprt("++ empty cap img......\n");
				return 0;
			}	
			
			if(!send_buf)
			{
				logprt("### Error: encoding.....\n");
				return 0;
			}

			//logprt("== set tx: jpegSize=%d, \n", jpegSize);
			udpx->set(send_buf, jpegSize, 1472);
			udpx->tx();

			
		}
	}

	else if(msgid == EM_CLOSE)
	{
		es_kill_timer(psys, TIMER_SEND);
		udpx->close();
		delete udpx;
		stop_cap_thread();
		close_scr_encoder();
		
	}
	
	return 0;
}
void* tx_thread(void* arg)
{
	es_open(SCX);
	if(txinfo.trans==0)
		es_register_msgproc(SCX, TxMsgProc);
	else	
		es_register_msgproc(SCX, TxUDPMsgProc);
	es_main(SCX);
	es_close(SCX);
	return NULL;
}



int RxMsgProc(es_t* psys, uint16_t msgid, uint16_t p1, uint32_t p2)
{
#define SOCKID_LISTEN 1
#define SOCKID_CHILD 2
	static sock_t lsock, rsock;
	static sock_t *psock;
	static unsigned int total_read;
	static unsigned int tstart, tend;
	static char* read_buf;

	
	static int exp_size=0, cur_size=0;


	int ret;
	
	if(msgid == EM_INIT)
	{
		logprt("== RX MsgProc Init....");
		gGPBuf = new CGetPutBuf(3, PKT_BUF_SIZE);
		start_render_thread();
		
		exp_size = cur_size = 0;

		total_read = 0;

		psock = &rsock;
		tstart = tend = 0;

		es_sock_open(psys, psock, KSOCK_TYPE_TCP, NULL, (void*)SOCKID_CHILD);
		ret = es_connect(psock, rxinfo.ip, rxinfo.port);
		logprt("== connecting, remote=%s, port=%d, ret=%d\n", rxinfo.ip, rxinfo.port, ret);


		read_buf = NULL;
		
		return ret;	
		
	}
	else if(msgid == EM_SOCKET)
	{	
		if(p2 == SOCKID_CHILD) // child socket
		{
			if(p1 == SOCK_EVENT_READ)
			{
				int rdcnt, reqsize;
				for(;;)
				{
					if(read_buf==NULL)
					{
						read_buf = (char*)gGPBuf->get_empty();
						reqsize = sizeof(int);
						if(!read_buf)
						{
							logprt("### read buf fail..\n");
							es_exit(psys);
							break;
						}
					}	
					
					if(cur_size< PKT_HDR_SIZE)
					{
						reqsize = PKT_HDR_SIZE - cur_size;
					}	
					else
						reqsize = exp_size - cur_size;
					

					
					if(reqsize == 0)
					{
						logprt("#### read size error,\n");
						exit(1);
						//es_exit(psys);
					}


					//reqsize = 150*1024; // test
					rdcnt = es_read(psock, read_buf+cur_size, reqsize);
					logprt("== rdcnt=%d, read_buf=%p, req_size=%d, total=%d, cur_size=%d, errno=%d\n", rdcnt, read_buf, reqsize, total_read, cur_size, errno);
					//gGPBuf->put_empty(read_buf);
					//return rdcnt; // test
					if(rdcnt==0)
					{
						logprt("== disconnected, err=%d\n", errno);
						es_tcp_close(psock);
						logprt("++++++++ total bytes = %d\n", total_read);
						es_exit(psys);
						break;
					}
					else if(rdcnt > 0)
					{
						total_read += rdcnt;
						cur_size += rdcnt;
						
						if(exp_size==0)
						{
							if(cur_size >= PKT_HDR_SIZE)
							{
								exp_size = ((int*)read_buf)[0];
								//logprt("== new packet start, size=%d\n", exp_size);	
							}
						}
						else
						{
							if(cur_size==exp_size) 
							{	
								//logprt("== end packet, \n");
								cur_size = exp_size = 0;
								gGPBuf->put_fill(read_buf);
								read_buf = NULL;
							}
							else if(cur_size > exp_size) 
							{
								logprt("### size recv error, cur_size=%d, exp_size=%d\n", cur_size, exp_size);
								exit(1);
							}
						}
						

					}
					else if(rdcnt < 0)
					{
						//logprt("== read error, rdcnt=%d, err=%d\n", rdcnt, errno);
						break;
					}
				}	
				return rdcnt;
			}
			else if(p1 == SOCK_EVENT_CONNECTED)
			{
				logprt("== recv connected.... start render samp....\n");
				start_render_samp();
				return 0;
			}
			else if(p1 == SOCK_EVENT_DISCONNECTED)
			{
				logprt("== recv disconnected ... end render samp.....\n");
				stop_render_samp();
				return 0;
			}
			
		}
	}
	else if(msgid == EM_CLOSE)
	{
		logprt("== RxMsgProc close.........\n");
		stop_render_thread();
		es_tcp_close(psock);
		
		delete gGPBuf;
	}
	return 0;
}


int RxUDPMsgProc(es_t* psys, uint16_t msgid, uint16_t p1, uint32_t p2)
{
	static char* read_buf;
	static CUdpXBuf* udprx;
	int ret;
	if(msgid == EM_INIT)
	{
		logprt("== RX MsgProc Init....");
		gGPBuf = new CGetPutBuf(3, PKT_BUF_SIZE);
		udprx = new CUdpXBuf(psys, 100*1024, 0, gGPBuf);

		logprt("== listen udp port=%d\n", rxinfo.port);
		udprx->listen(rxinfo.port);
		start_render_thread();
		usleep(1000*5);
		start_render_samp();

		return 0;	
		
	}
	else if(msgid == EM_CLOSE)
	{
		logprt("== RxMsgProc close.........\n");
		stop_render_thread();
		udprx->close();
		delete udprx;
		delete gGPBuf;
		
	}
	return 0;
}

void* rx_thread(void* arg)
{
	es_open(RCX);
	if(rxinfo.trans == 0)
		es_register_msgproc(RCX,  RxMsgProc);
	else	
		es_register_msgproc(RCX,  RxUDPMsgProc);
	es_main(RCX);
	es_close(RCX);
	return NULL;
}


void setMirrorCfg(int w, int h)
{
	txinfo.imgw = w;
	txinfo.imgh = h;

	rxinfo.imgw = w;
	rxinfo.imgh = h;
}
